Which issue does this PR close?

• Relates to datafusion.execution.collect_statistics on wide tables #21624 (datafusion.execution.collect_statistics on wide tables).

Rationale for this change

Today, statistics flow through DataFusion as an all-or-nothing dense Statistics struct: collect_statistics=true reads parquet thrift footers for every column of every file, allocates a Vec<ColumnStatistics> of length num_columns per file, and stores it whether the query references those columns or not. On wide tables that's a lot of memory and IO that the query may not need.

This is even worse for 3rd party TableProvider implementations that may store statistics in an external catalog or Parquet files in object store (think Delta/Iceberg, a Postgres backed table, etc). These implementations could efficiently pull statistics for 1-2 columns but are forced to pull all statistics since they don't know which ones DataFusion wants leading to a lot of wasted work.

This PR proposes a small handshake on the existing scan_with_args API that lets a caller ask a TableProvider for specific stats by name, and lets the provider answer only what it can deliver cheaply. The shape is intentionally minimal — enough to unblock memory and third-party wins on its own, with room to grow toward sketches, histograms, and selectivity stats later.

What changes are included in this PR?

The change is split into two stacked commits.

My goal was to implement the APIs, statistics collection and FilePruner as the first consumer. Porting all other consumers is a larger piece of work that I think should be done across multiple PRs.

Commit 1: catalog: query-aware statistics requests via ScanArgs / ScanResult

The API surface itself.

New types in datafusion-expr-common::statistics

pub enum StatisticsRequest {
    Min(Column), Max(Column),
    NullCount(Column), DistinctCount(Column),
    Sum(Column), ByteSize(Column),
    RowCount, TotalByteSize,
}

pub enum StatisticsValue {
    Scalar(Precision<ScalarValue>),
    Distribution(Box<Distribution>),  // typed; reuses existing enum
    Absent,
}

pub type SatisfiedStatistics =
    HashMap<StatisticsRequest, StatisticsValue>;

The variants of StatisticsRequest mirror the fields of Statistics
/ ColumnStatistics, so a provider that already populates one can
answer the other trivially. Whether a value is exact or estimated
travels in the Precision wrapper, not in the request kind itself.

ScanArgs / ScanResult extension

ScanArgs::with_statistics_requests(requests: &[StatisticsRequest])
ScanArgs::statistics_requests() -> &[StatisticsRequest]
ScanResult::with_statistics(statistics: Vec<StatisticsValue>)
ScanResult::statistics() -> &[StatisticsValue]

The contract: "answer what's free, leave the rest as Absent."

Per-file sparse stats

PartitionedFile.satisfied_stats: Option<Arc<SatisfiedStatistics>>
PartitionedFile::with_satisfied_stats(...)

Memory scales with what was asked for, not with table width.

FilePruner consumes the sparse map

FilePruner learns to dispatch between the existing PrunableStatistics
(dense) and a new SparseFilePruningStats adapter that does direct
request-keyed lookups against the sparse map. No densify-then-throw-away.

ListingTable answers from footer metadata

When a caller passes with_statistics_requests(...), scan_with_args
populates ScanResult.statistics from the merged dense Statistics it
already touched. Min/Max/NullCount/RowCount/TotalByteSize come back as
the Precision value the format produced. DistinctCount/Sum/ByteSize
come back as Absent for parquet — those aren't in thrift footers.

Commit 2: optimizer: derive StatisticsRequests from logical plan

The producer side: now the planner actually asks.

• TableScan gains statistics_requests: Vec<StatisticsRequest> and a
  with_statistics_requests builder.

• New RequestStatistics OptimizerRule runs last in the default
  pipeline. Walks the optimized plan once, derives:

  node	requests
  Sort	Min / Max / NullCount on each sort key
  Filter	Min / Max / NullCount / DistinctCount on referenced cols
  Join	DistinctCount / NullCount on join keys (both sides)
  always	RowCount per scan

  Idempotent; never reshapes the plan, only annotates TableScans.

• DefaultPhysicalPlanner reads scan.statistics_requests and threads
  them into ScanArgs::with_statistics_requests when calling
  provider.scan_with_args.

End to end: SQL → optimizer → TableScan.statistics_requests →
physical planner → provider.scan_with_args(ScanArgs{statistics_requests, ..})
→ provider returns whatever it can on ScanResult.statistics →
caller (today: nothing built-in; tomorrow: a layered stats helper or
optimizer rule) consumes.

Future work

Advanced stats not available from Parquet footers

In designing this change I wanted to keep in mind some of the other goals discussed in #21624.
In particular, I think we should develop a story for being able to collect more advanced planning statistics ad-hoc when they are not available.

I was inspired by Floe SQL's CMU talk.
The idea would be:

1. From the logical plan we derive statistics that we think we'll need.
2. We pass them into the TableProvider implementations (this PR) and they answer what they can. E.g. if your metadata store keeps track of most common values for a column, congrats! Populate it from there. But if you have a bunch of random Parquet files just leave it blank.
3. Now we have a physical plan + a bunch of missing stats we want. Maybe we augment this by asking the physical optimizers what stats they want, etc. But now we do something similar to what Floe does: we run a mini query to sample data and collect statistics. Some things that make this tractable which I've prototyped:
   a) File -> Row group -> Row Chunk sampling. Basically a semi-efficient way to say "get me a random sample of 1% of the data". I started POC work on this in feat: TABLESAMPLE SYSTEM end-to-end + row-group / row sampling on ParquetSource #22000 which also stands on it's own.
   b) Time and row bounded scanning. This sampled data is a stream that flows into aggregators that compute stats. We put a time (say 150ms) and row number (say 100k) limit on this stream. This guarantees that our worst case cost is constant. If we bail out early we can emit our best estimate up to this point or leave the stats unpopulated.
   We can also play all sorts of games with timing, e.g. we can start the query and the stats collection and race them. If the query finished before the stats collection (possible if e.g. there's a single file and opening the file is the dominating cost) then just cancel stats collection. If stats collection finishes first we can re-optimize the plan and if the new plan is "better" enough that we think it's worth the price we cancel and re-run the query. I imagine this might be a good config knob.

I also imagine there might be some hooks in the implementation of how to collect those stats. A default one we bundle might do the sampling from parquet files, but I can imagine users may want to hook in some catalog of stats they have if they deem it too expensive to run as part of TableProvider::scan_with_args, e.g. if they store a pre-sampled set of rows or just advanced statistics like sketches (that can't be represented by Parquet's simplistic stats model) as an embedded index.

Expression based statistics

Discussed in #20871. We represent struct field access in datafusion as an expression get_field(col, 'field'). It is not a column. Our existing statistics system cannot represent statistics for a struct field (only the top level struct, which is meaningless most of the time) even though Parquet carries stats for leaf columns of nested types.

This proposal would make it much easier to implement this, we would need to expand StatisticsRequest with Min(Expr)/Max(Expr), etc. We could fold Column into there or keep it as a special case. This mirrors what engines like Postgres can do (you can tell it to collect and store stats on expressions). Then a PartitionedFile can have file.stats.get(Min(Expr(get_field(col, 'field'))) -> StatisticsResult::Scalar(...)` (pseudo code to avoid UDF invocation ceremony).

Relationship to other PRs

There is a lot of work being done around statistics, I will mention just the ones that stand out in my mind.

This PR deals with 2 things:

1. How do we decide and represent what statistics to collect?
2. How do we get those statistics from metadata stores or our data (for Parquet)?

It does not deal with:

1. Using those stats for pruning.
2. Propagating stats through expressions or execution plans.

#19609

This PR deals with statistics propagation and statistics representation.
I don't think there is much conflict in propagation (the current PR does not deal with that) but the representation piece is interesting. I can see a world where the HashMap<StatisticsRequest, StatisticsResponse> is instead a RecordBatch + a wrapper that offers an API like stats.get([Min("column"), Max("other column")]) but knows how to look that up in the RecordBatch and produce a new RecordBatch as a result that can be used for vectorized operations on statics.

Right now I am more concerned with the IO / memory cost of stats and having APIs to be able to understand what stats are needed and request and represent advanced stats more so than I am with vectorized execution on stats. In my timings from #21968 (comment) stats computation (e.g. PruningPredicate) is not a major issue, it's stats collection that costs us a lot. I've seen similar things in our system where we store stats in Parquet files. I also have some trouble grasping how we are going to fit and what the advantages are of storing things like sketches or histograms in arrow arrays.

#21122

This work also deals with statistics propagation. It is orthogonal, but there is some overlap worth discussing. In particular in the section above about struct field stats I proposed we allow statistics requests to contain expressions. It then becomes unclear if, in the case of JOIN t2 ON coalesce(t1.col, '') = t2.col if we should use the APIs in this PR to request StatisticsRequest::Min(Expr(coalesce(t1.col, ''))) or if we should request stats on t1.col and use something like #21122 to propagate them.

I think this won't be that hard to resolve: the get_field functions are already special via

Maybe we can use that existing API? Or we can come up with something new. But ultimately it seems resolvable to split expressions into ones we ask the data source for vs. derive.

Are there any user-facing changes?

API additions only, all opt-in:

• ScanArgs / ScanResult gain new fields with Default-friendly initializers; existing callers that don't use the new builders see no change.
• FilePruner's field-type change is a private internal field.

The only source-level break: PartitionedFile gains a new pub satisfied_stats: Option<...> field. Callers using PartitionedFile::new / From<ObjectMeta> / the existing builders are unaffected. Direct struct literals (uncommon, none in-tree) need to add satisfied_stats: None or migrate to the new with_satisfied_stats builder.